package com.gagakuai.concurrent.locks;

import com.gagakuai.concurrent.BlockingQueue;

import java.io.Serializable;
import java.lang.ref.WeakReference;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

/*
 * 顺序有界（容量固定） 循环阻塞队列，线程安全（锁）
 * <p>
 * 注：该循环队列是通过count域的值来识别队空还是队满的
 *
 * @author: houhong
 */
public class ArrayBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, Serializable {

    /*
     * 存储队列中的元素
     */
    final Object[] items;

    /*
     * 队头
     */
    int takeIndex;
    /*
     * 队尾 为null 等待接收元素
     */
    int putIndex;
    /*
     * 队中元素个数
     */
    int count;

    /*
     * 队列锁
     */
    final ReentrantLock lock;

    /*
     * 出队条件
     */
    private final Condition notEmpty;

    /*
     * 入队条件
     */
    private final Condition notFull;

    transient Itrs itrs;

    /*==========构造器============================================================*/


    public ArrayBlockingQueue(int capacity) {
        this(capacity, false);
    }


    public ArrayBlockingQueue(int capacity, boolean fair) {

        if (capacity <= 0) {
            throw new IllegalArgumentException("capacity should not be null");
        }
        this.items = new Object[capacity];
        this.lock = new ReentrantLock(fair);
        this.notEmpty = lock.newCondition();
        this.notFull = lock.newCondition();

    }

    /*
     * 用指定容器中的数据初始化阻塞队列
     */
    public ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c) {

        this(capacity, fair);

        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            final Object[] items = this.items;

            int index = 0;
            for (E e : c) {
                items[index++] = Objects.requireNonNull(e);
            }
            //记录总数
            count = index;
            //设置队尾索引
            putIndex = (index == capacity) ? index : 0;
        } catch (ArrayIndexOutOfBoundsException ex) {
            throw new IllegalArgumentException("this  size of collection should not be more than capcity");
        } finally {
            lock.unlock();
        }

    }


    /*==========构造器结束============================================================*/




    /*==========入队开始============================================================*/


    /*
     * 入队，线程安全，队满时线程被阻塞
     */
    @Override
    public void put(E e) throws InterruptedException {

        Objects.requireNonNull(e);

        final ReentrantLock lock = this.lock;

        // 申请独占锁，不允许阻塞带有中断标记的线程
        lock.lockInterruptibly();
        try {
            // 如果队列满了，需要阻塞“入队”线程
            while (count == items.length) {
                /*
                *  通过条件队列，将当前线程包装成node 节点放入到 条件队列中
                */
                notFull.await();
            }
            //入队
            enqueue(e);
        } finally {
            lock.unlock();
        }

    }


    /*
     * 入队，线程安全，队满时阻塞一段时间，如果在指定的时间内没有机会插入元素，则返回false
     */
    @Override
    public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {


        Objects.requireNonNull(e);

        long timeOutNaos = unit.toNanos(timeout);

        // 申请独占锁，不允许阻塞带有中断标记的线程
        lock.lockInterruptibly();

        // 当队列满了
        try {
            while (count == items.length) {
                //超时
                if (timeOutNaos <= 0L) {
                    return false;
                }
                //阻塞一段时间
                timeOutNaos = notFull.awaitNaos(timeOutNaos);
            }
            //入队
            enqueue(e);
            return false;
        } finally {
            lock.unlock();
        }
    }


    /*
     * 入队 非线程安全,用于内部使用
     */
    private void enqueue(E e) {

        final Object[] items = this.items;

        //元素入队
        items[putIndex] = e;

        //当队列满了以后，设置putIndex  = 0;
        if (++putIndex == items.length) {
            putIndex = 0;
        }

        // 元素数量增加一个
        count++;

        // 唤醒出队线程
        notEmpty.signal();
    }













    /*==========入队结束============================================================*/



    /*==========出队开始============================================================*/

    /*
     * 出队，线程安全，在队空时候被阻塞
     */
    @Override
    public E take() throws InterruptedException {

        final ReentrantLock lock = this.lock;

        // 申请独占锁，禁止中断标记
        lock.lockInterruptibly();
        try {
            //当队列为空的时候，需要阻塞出队线程
            while (count == 0) {
                notEmpty.await();
            }
            //出队
            return dequeue();
        } finally {
            lock.unlock();
        }
    }

    /*
     * 出队,非线程安全，队列内部使用
     */
    private E dequeue() {

        final Object[] items = this.items;

        E item = (E) items[takeIndex];

        // 防止内存泄露
        items[takeIndex] = null;

        // 游标递增（轮转）
        if (++takeIndex == items.length) {
            takeIndex = 0;
        }

        count--;

        if (itrs != null) {
            itrs.elementDequeued();
        }

        //唤醒入队线程
        notFull.signal();

        return item;
    }

    private class Itr implements Iterator<E> {

        private static final int NONE = -1;

        private static final int REMOVED = -2;

        private static final int DETACHED = -3;

        private int cursor;
        private E nextItem;

        private int nextIndex;

        private E lastItem;

        private int lastRet;

        private int prevTakeIndex;

        private int prevCycles;

        Itr() {
            lastRet = NONE;
            final ReentrantLock lock = ArrayBlockingQueue.this.lock;
            lock.lock();
            try {
                if (count == 0) {
                    // assert itrs == null;
                    cursor = NONE;
                    nextIndex = NONE;
                    prevTakeIndex = DETACHED;
                } else {
                    final int takeIndex = ArrayBlockingQueue.this.takeIndex;

                    prevTakeIndex = takeIndex;

                    // 初始化为队头元素
                    nextItem = itemAt(nextIndex = takeIndex);
                    cursor = incCursor(takeIndex);
                    if (itrs == null) {
                        itrs = new Itrs(this);
                    } else {
                        itrs.register(this); // in this order
                        itrs.doSomeSweeping(false);
                    }
                    prevCycles = itrs.cycles;
                    // assert takeIndex >= 0;
                    // assert prevTakeIndex == takeIndex;
                    // assert nextIndex >= 0;
                    // assert nextItem != null;
                }
            } finally {
                lock.unlock();
            }
        }


        @Override
        public boolean hasNext() {
            if (nextItem != null) {
                return true;
            }
            noNext();
            return false;
        }

        @Override
        public E next() {
            final E e = nextItem;
            if (e == null) {
                throw new NoSuchElementException();
            }
            final ReentrantLock lock = ArrayBlockingQueue.this.lock;
            lock.lock();
            try {
                if (!isDetached()) {
                    incorporateDequeues();
                }

                lastRet = nextIndex;
                final int cursor = this.cursor;
                if (cursor >= 0) {
                    nextItem = itemAt(nextIndex = cursor);

                    this.cursor = incCursor(cursor);
                } else {
                    nextIndex = NONE;
                    nextItem = null;
                    if (lastRet == REMOVED) {
                        detach();
                    }
                }
            } finally {
                lock.unlock();
            }
            return e;
        }

        @Override
        public void forEachRemaining(Consumer<? super E> action) {
            Objects.requireNonNull(action);
            final ReentrantLock lock = ArrayBlockingQueue.this.lock;
            lock.lock();
            try {
                final E e = nextItem;
                if (e == null) {
                    return;
                }

                if (!isDetached()) {
                    incorporateDequeues();
                }

                action.accept(e);

                if (isDetached() || cursor < 0) {
                    return;
                }

                final Object[] items = ArrayBlockingQueue.this.items;
                for (int i = cursor, end = putIndex, to = (i < end) ? end : items.length; ; i = 0, to = end) {
                    for (; i < to; i++) {
                        action.accept(itemAt(items, i));
                    }
                    if (to == end) {
                        break;
                    }
                }
            } finally {
                // Calling forEachRemaining is a strong hint that this
                // iteration is surely over; supporting remove() after
                // forEachRemaining() is more trouble than it's worth
                cursor = nextIndex = lastRet = NONE;
                nextItem = lastItem = null;
                detach();
                lock.unlock();
            }
        }

        @Override
        public void remove() {
            final ReentrantLock lock = ArrayBlockingQueue.this.lock;
            lock.lock();
            // assert lock.getHoldCount() == 1;
            try {
                if (!isDetached()) {
                    incorporateDequeues(); // might update lastRet or detach
                }
                final int lastRet = this.lastRet;
                this.lastRet = NONE;
                if (lastRet >= 0) {
                    if (!isDetached()) {
                        // 移除指定索引处的元素，非线程安全
                        removeAt(lastRet);
                    } else {
                        final E lastItem = this.lastItem;
                        // assert lastItem != null;
                        this.lastItem = null;
                        if (itemAt(lastRet) == lastItem) {
                            // 移除指定索引处的元素，非线程安全
                            removeAt(lastRet);
                        }
                    }
                } else if (lastRet == NONE) {
                    throw new IllegalStateException();
                }
                if (cursor < 0 && nextIndex < 0) {
                    detach();
                }
            } finally {
                lock.unlock();
                // assert lastRet == NONE;
                // assert lastItem == null;
            }
        }

        boolean isDetached() {
            // assert lock.isHeldByCurrentThread();
            return prevTakeIndex < 0;
        }


        void shutdown() {
            cursor = NONE;
            if (nextIndex >= 0) {
                nextIndex = REMOVED;
            }
            if (lastRet >= 0) {
                lastRet = REMOVED;
                lastItem = null;
            }
            prevTakeIndex = DETACHED;
        }


        boolean removedAt(int removedIndex) {

            if (isDetached()) {
                return true;
            }

            final int takeIndex = ArrayBlockingQueue.this.takeIndex;
            final int prevTakeIndex = this.prevTakeIndex;
            final int len = items.length;

            final int removedDistance = len * (itrs.cycles - this.prevCycles + ((removedIndex < takeIndex) ? 1 : 0)) + (removedIndex - prevTakeIndex);

            int cursor = this.cursor;
            if (cursor >= 0) {
                int x = distance(cursor, prevTakeIndex, len);
                if (x == removedDistance) {
                    if (cursor == putIndex) {
                        this.cursor = cursor = NONE;
                    }
                } else if (x > removedDistance) {

                    this.cursor = cursor = dec(cursor, len);
                }
            }
            int lastRet = this.lastRet;
            if (lastRet >= 0) {
                int x = distance(lastRet, prevTakeIndex, len);
                if (x == removedDistance) {
                    this.lastRet = lastRet = REMOVED;
                } else if (x > removedDistance) {
                    this.lastRet = lastRet = dec(lastRet, len);
                }
            }
            int nextIndex = this.nextIndex;
            if (nextIndex >= 0) {
                int x = distance(nextIndex, prevTakeIndex, len);
                if (x == removedDistance) {
                    this.nextIndex = nextIndex = REMOVED;
                } else if (x > removedDistance) {
                    this.nextIndex = nextIndex = dec(nextIndex, len);
                }
            }
            if (cursor < 0 && nextIndex < 0 && lastRet < 0) {
                this.prevTakeIndex = DETACHED;
                return true;
            }
            return false;
        }


        boolean takeIndexWrapped() {
            // assert lock.isHeldByCurrentThread();
            if (isDetached()) {
                return true;
            }
            if (itrs.cycles - prevCycles > 1) {

                shutdown();
                return true;
            }
            return false;
        }

        private int incCursor(int index) {
            // assert lock.isHeldByCurrentThread();
            if (++index == items.length) {
                index = 0;
            }
            if (index == putIndex) {
                index = NONE;
            }
            return index;
        }


        private boolean invalidated(int index, int prevTakeIndex, long dequeues, int length) {
            if (index < 0) {
                return false;
            }
            int distance = index - prevTakeIndex;
            if (distance < 0) {
                distance += length;
            }
            return dequeues > distance;
        }


        private void incorporateDequeues() {


            final int cycles = itrs.cycles;
            final int prevCycles = this.prevCycles;

            final int takeIndex = ArrayBlockingQueue.this.takeIndex;
            final int prevTakeIndex = this.prevTakeIndex;

            if (cycles != prevCycles || takeIndex != prevTakeIndex) {
                final int len = items.length;

                long dequeues = (long) (cycles - prevCycles) * len + (takeIndex - prevTakeIndex);


                if (invalidated(lastRet, prevTakeIndex, dequeues, len)) {
                    lastRet = REMOVED;
                }
                if (invalidated(nextIndex, prevTakeIndex, dequeues, len)) {
                    nextIndex = REMOVED;
                }
                if (invalidated(cursor, prevTakeIndex, dequeues, len)) {
                    cursor = takeIndex;
                }

                if (cursor < 0 && nextIndex < 0 && lastRet < 0) {
                    detach();
                } else {
                    this.prevCycles = cycles;
                    this.prevTakeIndex = takeIndex;
                }
            }
        }


        private void detach() {

            if (prevTakeIndex >= 0) {

                prevTakeIndex = DETACHED;

                itrs.doSomeSweeping(true);
            }
        }

        private void noNext() {
            final ReentrantLock lock = ArrayBlockingQueue.this.lock;
            lock.lock();
            try {

                if (!isDetached()) {

                    incorporateDequeues();
                    if (lastRet >= 0) {
                        lastItem = itemAt(lastRet);

                        detach();
                    }
                }

            } finally {
                lock.unlock();
            }
        }

        private int distance(int index, int prevTakeIndex, int length) {
            int distance = index - prevTakeIndex;
            if (distance < 0) {
                distance += length;
            }
            return distance;
        }

    }

    /*
     * 迭代器
     */
    class Itrs {

        int cycles;

        private static final int SHORT_SWEEP_PROBES = 4;
        private static final int LONG_SWEEP_PROBES = 16;

        /*
         * Linked list of weak iterator references
         */
        private Node head;

        /*
         * Used to expunge stale iterators
         */
        private Node sweeper;

        Itrs(Itr initial) {
            register(initial);
        }

        void elementDequeued() {
            // assert lock.isHeldByCurrentThread();
            if (count == 0) {
                queueIsEmpty();
            } else if (takeIndex == 0) {
                takeIndexWrapped();
            }

        }

        void register(Itr itr) {
            // assert lock.isHeldByCurrentThread();
            head = new Node(itr, head);
        }


        void doSomeSweeping(boolean tryHarder) {
            // assert lock.isHeldByCurrentThread();
            // assert head != null;
            int probes = tryHarder ? LONG_SWEEP_PROBES : SHORT_SWEEP_PROBES;
            Node o, p;
            final Node sweeper = this.sweeper;
            boolean passedGo;   // to limit search to one full sweep

            if (sweeper == null) {
                o = null;
                p = head;
                passedGo = true;
            } else {
                o = sweeper;
                p = o.next;
                passedGo = false;
            }

            for (; probes > 0; probes--) {
                if (p == null) {
                    if (passedGo) {
                        break;
                    }
                    o = null;
                    p = head;
                    passedGo = true;
                }
                final Itr it = p.get();
                final Node next = p.next;
                if (it == null || it.isDetached()) {
                    // found a discarded/exhausted iterator
                    probes = LONG_SWEEP_PROBES; // "try harder"
                    // unlink p
                    p.clear();
                    p.next = null;
                    if (o == null) {
                        head = next;
                        if (next == null) {
                            // We've run out of iterators to track; retire
                            itrs = null;
                            return;
                        }
                    } else {
                        o.next = next;
                    }
                } else {
                    o = p;
                }
                p = next;
            }

            this.sweeper = (p == null) ? null : o;
        }

        void queueIsEmpty() {
            // assert lock.isHeldByCurrentThread();
            for (Node p = head; p != null; p = p.next) {
                Itr it = p.get();
                if (it != null) {
                    p.clear();
                    it.shutdown();
                }
            }
            head = null;
            itrs = null;
        }

        void removedAt(int removedIndex) {
            for(Node o = null, p = head; p != null; ) {
                final Itr it = p.get();
                final Node next = p.next;
                if(it == null || it.removedAt(removedIndex)) {
                    // unlink p
                    // assert it == null || it.isDetached();
                    p.clear();
                    p.next = null;
                    if(o == null) {
                        head = next;
                    } else {
                        o.next = next;
                    }
                } else {
                    o = p;
                }
                p = next;
            }
            if(head == null)
            {
                itrs = null;
            }
        }
        void takeIndexWrapped() {
            // assert lock.isHeldByCurrentThread();
            cycles++;
            for(Node o = null, p = head; p != null; ) {
                final Itr it = p.get();
                final Node next = p.next;
                if(it == null || it.takeIndexWrapped()) {
                    // unlink p
                    // assert it == null || it.isDetached();
                    p.clear();
                    p.next = null;
                    if(o == null) {
                        head = next;
                    } else {
                        o.next = next;
                    }
                } else {
                    o = p;
                }
                p = next;
            }
            if(head == null)   // no more iterators to track
            {
                itrs = null;
            }
        }

        private class Node extends WeakReference<Itr> {
            Node next;

            Node(Itr iterator, Node next) {
                super(iterator);
                this.next = next;
            }


        }
    }


    @Override
    public boolean add() throws InterruptedException {
        return false;
    }


    @Override
    public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        return null;
    }

    @Override
    public int drainTo(Collection<? super E> c) {
        return 0;
    }

    @Override
    public int drainTo(Collection<? super E> c, int maxElements) {
        return 0;
    }

    @Override
    public int remainingCapacity() {
        return 0;
    }

    @Override
    public Iterator<E> iterator() {
        return null;
    }

    @Override
    public int size() {
        return 0;
    }

    @Override
    public boolean offer(E e) {
        return false;
    }

    @Override
    public E poll() {
        return null;
    }

    @Override
    public E peek() {
        return null;
    }

    static final int dec(int i, int modulus) {
        if (--i < 0) {
            i = modulus - 1;
        }
        return i;
    }

    final E itemAt(int i) {
        return (E) items[i];
    }

    static <E> E itemAt(Object[] items, int i) {
        return (E) items[i];
    }


    /*
     * 移除指定索引处的元素，非线程安全，仅限内部使用
     */
    void removeAt(final int removeIndex) {

        final Object[] items = this.items;

        // 移除队头元素
        if (removeIndex == takeIndex) {
            // removing front item; just advance
            items[takeIndex] = null;
            if (++takeIndex == items.length) {
                takeIndex = 0;
            }
            count--;
            if (itrs != null) {
                itrs.elementDequeued();
            }

            // 移除非队头元素（需要移动元素）
        } else {

            for (int i = removeIndex, putIndex = this.putIndex; ; ) {
                int pred = i;
                if (++i == items.length) {
                    i = 0;
                }
                if (i == putIndex) {
                    items[pred] = null;
                    this.putIndex = pred;
                    break;
                }
                items[pred] = items[i];
            }
            count--;
            if (itrs != null) {
                itrs.removedAt(removeIndex);
            }
        }
        notFull.signal();
    }
}